package com.chen.flinkdemo.source;

import com.chen.flinkdemo.Access;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * @author: cy
 * @date: 2024/8/29 23:12
 * @description:  SourceFunction 的并行量默认是1
 */


public class AccessSource  implements ParallelSourceFunction<Access> {
    boolean running = true;

    @Override
    public void run(SourceContext<Access> ctx) throws Exception {
        Random random = new Random();
        String[] domains = {"imooc.com", "a.com","b.com"};


        while (running) {

            // 生成数据
            for (int i = 0; i < 10 ; i++) {
                Access access = new Access();
                access.setTime(1234567L);
                access.setDomain(domains[random.nextInt(domains.length)]);
                access.setTraffic(random.nextDouble() + 1000);

                ctx.collect(access);
            }

            Thread.sleep(5000);
        }
    }


    @Override
    public void cancel() {
        running = false;
    }
}
